package com.rocketmq.one.message;

import com.rocketmq.one.domain.MsgModul;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

public class OrderMessage {

    List<MsgModul> moduls = Arrays.asList(
            new MsgModul(101,1001,"下单"),
            new MsgModul(101,1001,"短信"),
            new MsgModul(101,1001,"支付"),
            new MsgModul(102,1002,"下单"),
            new MsgModul(102,1002,"短信"),
            new MsgModul(102,1002,"支付")
    );

    @Test
    public void orderlyProduce() throws Exception {
        // 获取生产者对象
        DefaultMQProducer producer = new DefaultMQProducer("orderlyProduce");
        // 设置服务对象
        producer.setNamesrvAddr("8.134.163.217:9876");
        // 开始
        producer.start();
        // 处理
        for (MsgModul m : moduls) {
            Message msg = new Message("TopicTest", m.toString().getBytes());
            producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message message, Object o) {
                    int index = (int) o % mqs.size();
                    return mqs.get(index);
                }
            }, m.getUserId());
        }
        producer.shutdown();

    }
    @Test
    public void consume() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderlyConsume");
        consumer.setNamesrvAddr("8.134.163.217:9876");
        consumer.subscribe("TopicTest", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("当前线程 --- "+Thread.currentThread().getId());
                System.out.println("消息内容："+new String(list.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();

    }
    @Test
    public void orderlyConsume() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderlyConsume");
        consumer.setNamesrvAddr("8.134.163.217:9876");
        consumer.subscribe("TopicTest", "*");
        // 并发模式 多线程消费
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                System.out.println("当前线程 --- "+Thread.currentThread().getId());
                System.out.println("消息内容："+new String(list.get(0).getBody()));
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.in.read();

    }
}
